package com.galeno.day09;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2614:34
 * TTL    Time ToLive
 * 设置状态的存活时间  默认情况下状态的数据会一直保存
 * 默认情况下,状态的数据会一直保存,但有的数据,以后就不在使用了,再存储就有些浪费资源
 * 可以设置状态的存活时间.超时 的状态,flink会进行清除
 *
 *
 * 需求,最近两小时成交金额(保留当前小时,和上一个小时的数据)
 */
public class ValueTTLDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //商品分类
        //商品分类,商品总金额
        //家具,1000
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.3", 9999);
        SingleOutputStreamOperator<Tuple2<String, Double>> tpStream = lines.map(new MapFunction<String, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple2.of(split[0], Double.parseDouble(split[1]));
            }
        });
        KeyedStream<Tuple2<String, Double>, String> keyedStream = tpStream.keyBy(x -> x.f0);
       // keyedStream.process();




    }

    /**
     * 家具 2021102515,5000
     */
    private static class IncomeHourlyFunction extends KeyedProcessFunction<String,Tuple2<String,Double>, Tuple3<String,String,Double>>{

        @Override
        public void open(Configuration parameters) throws Exception {
            MapStateDescriptor<String, Double> stateDescriptor = new MapStateDescriptor<>("hourly-state", String.class, Double.class);
            //设置状态描述器设置TTL
            //StateTtlConfig builder = StateTtlConfig.newBuilder(Time.minutes(2));//为了效果演示
            StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.minutes(10))
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)//默认的,当创建和修改该key对应的value就会对TTL重新计时.
                   // .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)//创建修改读取都会重新计时
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)//设置状态的可见性默认,如果超时就不再返回了
                    //.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)//即使超时了,但是还没有被清除掉,也可以使用
                    .build();

             stateDescriptor.enableTimeToLive(ttlConfig);



        }

        @Override
        public void processElement(Tuple2<String, Double> value, Context ctx, Collector<Tuple3<String, String, Double>> out) throws Exception {

        }
    }
}

